package cn.lingyangwl.agile.member.listener;

import cn.lingyangwl.agile.member.constant.AccountFromType;
import cn.lingyangwl.agile.member.constant.AccountType;
import cn.lingyangwl.agile.member.model.entity.MemberLevel;
import cn.lingyangwl.agile.member.model.entity.MemberUser;
import cn.lingyangwl.agile.member.model.mq.MemberUserLevelUpgradeMQ;
import cn.lingyangwl.agile.member.model.rqrs.account_detail.AccountDetailSaveReq;
import cn.lingyangwl.agile.member.service.AccountDetailService;
import cn.lingyangwl.agile.member.service.MemberLevelService;
import cn.lingyangwl.agile.member.service.MemberUserService;
import cn.lingyangwl.framework.mq.base.manager.MqManager;
import cn.lingyangwl.framework.tool.core.exception.BizException;
import com.alibaba.fastjson2.JSON;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Objects;
import java.util.Optional;

/**
 * @author shenguangyang
 */
@Slf4j
@Component
@RabbitListener(bindings = {
        @QueueBinding(
                value = @Queue(value = MemberUserLevelUpgradeMQ.QUEUE, durable = "true"), key = MemberUserLevelUpgradeMQ.KEY,
                exchange = @Exchange(value = MemberUserLevelUpgradeMQ.EXCHANGE, type = ExchangeTypes.TOPIC))
})
public class UserLevelUpgradeListener {
    @Resource
    private MqManager mqManager;
    @Resource
    private AccountDetailService accountDetailService;
    @Resource
    private MemberUserService userService;
    @Resource
    private MemberLevelService levelService;

    @RabbitHandler
    public void onMessage(String data, Channel channel, Message message) throws Exception {
        MemberUserLevelUpgradeMQ.MsgPayload msgPayload = MemberUserLevelUpgradeMQ.parse(data, MemberUserLevelUpgradeMQ.MsgPayload.class);
        try {
            if (mqManager.isConsumed(msgPayload.getMsgId())) {
                log.warn("message {} are repeatedly consumed, do not perform business", msgPayload.getMsgId());
                return;
            }
            MemberUser user = userService.getById(msgPayload.getUserId());
            if (Objects.isNull(user)) {
                return;
            }

            MemberLevel upgradeLevel = levelService.getById(msgPayload.getUpgradeLevelId());
            Integer sendPoint = Optional.ofNullable(upgradeLevel.getSendPoint()).orElse(0);

            // 发放升级奖励
            AccountDetailSaveReq saveReq = AccountDetailSaveReq.builder()
                    .accountData(new BigDecimal(sendPoint)).accountType(AccountType.POINT)
                    .user(user).fromType(AccountFromType.UPGRADE).build();
            accountDetailService.saveAccountDetail(saveReq);
            log.info("消费成功: {}", JSON.toJSONString(data));
        } catch (Exception e) {
            log.error("消费失败: {}, msg: {}", e.getMessage(), JSON.toJSONString(data));
            if (e instanceof NullPointerException) {
                log.error("error: ", e);
            } else if (e instanceof BizException) {
                throw e;
            }
            throw new BizException(-1, "mq消费错误");
        } finally {
            mqManager.markConsumed(msgPayload.getMsgId());

        }
    }
}
